

package mvcc

import (
	"bytes"
	"errors"
	"sync"

	"go.etcd.io/etcd/api/v3/mvccpb"
)

// AutoWatchID是在WatchStream中传递的观察者ID。当没有
// 用户提供的ID可用时进行观察。如果通过，将自动分配一个ID。
const AutoWatchID WatchID = 0

var (
	ErrWatcherNotExist    = errors.New("mvcc: watcher does not exist")
	ErrEmptyWatcherRange  = errors.New("mvcc: watcher range is empty")
	ErrWatcherDuplicateID = errors.New("mvcc: duplicate watch ID provided on the WatchStream")
)

type WatchID int64

// FilterFunc如果给定的事件应该被过滤掉，则返回true。
type FilterFunc func(e mvccpb.Event) bool

// watchableStore 实现了 Watchable 接口，该接
// 口中定义了 NewWatchStream（）方法，该方法会新建并返回 watchStream 实例 。结构体 watchStream
// 实现了 WatchStream 接口
type WatchStream interface {

	Watch(id WatchID, key, end []byte, startRev int64, fcs ...FilterFunc) (WatchID, error)

	Chan() <-chan WatchResponse

	RequestProgress(id WatchID)

	Cancel(id WatchID) error

	Close()

	Rev() int64
}

// WatchRespons巳表示一次响应，其中封装了多个Event实例
type WatchResponse struct {
	// 被触发的watcher实例的唯一标识。
	WatchID WatchID
	// 触发对应watcher事件集合。
	Events []mvccpb.Event
	// 当前watchResponse实例创建时对应的revision值
	Revision int64
	// 如果因为压缩操作对应watcher实例被取消了，则该字段设置为压缩操作对应的revision
	CompactRevision int64
}

// watchStream包含共享
// 一个流式chan以发送监视的事件和其他控制事件的监视程序集合。
type watchStream struct {
	// 用来记录关联的 watchableStore 实例。
	watchable watchable
	// 通过该watchStream实例创建的watcher实例在被触发时，都会将Event事件写入该通道中。
	ch        chan WatchResponse
	mu sync.Mutex // 保护其下的字段
	// 在当前 watchStream 实例中添加 watcher 实例时，会为其分
	// 配唯一标识，该字段就用来生成该唯一标识。
	nextID   WatchID
	closed   bool
	// 该字段记录唯一标识与取消对应 watcher 的回调 函数之间的映射关系。
	cancels  map[WatchID]cancelFunc
	// 该字段记录唯一标识与对应 watcher 实例之间的映射关系。
	watchers map[WatchID]*watcher
}

// watchStream.Watch（） 方法中会创建 watcher 实例监昕指定的 Key （或是范围监听［ key, end)),
// 其中 startRev 参数则指定了该 watcher 实例监昕的起始 revision ，如果 startRev 参数小于等于 0,
// 则表示从当前 revision 开始监听 ， fcs 参数则是前面介绍的 Event 事件过滤器。
func (ws *watchStream) Watch(id WatchID, key, end []byte, startRev int64, fcs ...FilterFunc) (WatchID, error) {
	// 检测参数的合法性，检测当前 watchStrearn 是否已经关闭
	if len(end) != 0 && bytes.Compare(key, end) != -1 {
		return -1, ErrEmptyWatcherRange
	}

	ws.mu.Lock()
	defer ws.mu.Unlock()
	if ws.closed {
		return -1, ErrEmptyWatcherRange
	}

	if id == AutoWatchID {
		for ws.watchers[ws.nextID] != nil {
			ws.nextID++
		}
		id = ws.nextID	// 生成新建 watcher 实例的唯一标识
		ws.nextID++
	} else if _, ok := ws.watchers[id]; ok {
		return -1, ErrWatcherDuplicateID
	}
	// 调用 watchableStore.watcher（）方法创建 watcher 实例，注意，新建的 watcher 实例共用了同一个通道
	w, c := ws.watchable.watch(key, end, startRev, id, ws.ch, fcs...)

	ws.cancels[id] = c	// 保存取消 watcher 实例时用到的回调函数
	ws.watchers[id] = w	// 记录新建的 watcher 实例
	return id, nil
}

func (ws *watchStream) Chan() <-chan WatchResponse {
	return ws.ch
}

// watchStream.Cancel（）方法中会查找并调用指定 watcher 实例对应的取消回调函数
func (ws *watchStream) Cancel(id WatchID) error {
	ws.mu.Lock()
	cancel, ok := ws.cancels[id]	// 根据 watcher 唯一标识，查找对应的取消回调函数
	w := ws.watchers[id]
	ok = ok && !ws.closed
	ws.mu.Unlock()

	if !ok {
		return ErrWatcherNotExist
	}
	cancel()	// 调用回调函数，取消 watcher 实例

	ws.mu.Lock()
	// 在取消之前不会删除该观察，以便如果调用Close（），它将等待取消。否则，Close（）可以在商店仍在发布事件时关闭观看频道。
	if ww := ws.watchers[id]; ww == w {
		delete(ws.cancels, id)	// 从 watchStrearn.cancels 中清理对应的取消回调函数
		delete(ws.watchers, id)	// 从 watchStrearn.watchers 字段中清理对应的 watcher 实告＇J
	}
	ws.mu.Unlock()

	return nil
}

func (ws *watchStream) Close() {
	ws.mu.Lock()
	defer ws.mu.Unlock()

	for _, cancel := range ws.cancels {
		cancel()
	}
	ws.closed = true
	close(ws.ch)
	watchStreamGauge.Dec()
}

func (ws *watchStream) Rev() int64 {
	ws.mu.Lock()
	defer ws.mu.Unlock()
	return ws.watchable.rev()
}

// watchStream.RequestProgress（）方法用来检测指定 watcher 的处理进度（ progress，即当前
// watcher 正在处理哪个 revision 中的更新操作） 。只有当 watcher 完全同步时，调用该方法才会创
// 建一个空的 watchResponse ，并写入 watcher.ch 通道中 ，
func (ws *watchStream) RequestProgress(id WatchID) {
	ws.mu.Lock()
	w, ok := ws.watchers[id]	// 从 watchStream.watchers 中查询指定的 watcher 实例
	ws.mu.Unlock()
	if !ok {
		return
	}
	ws.watchable.progress(w)	// 调用 watchableStore.progress （）方法
}
